package com.example.service.message.service;

import com.alibaba.fastjson.JSONObject;
import com.example.common.ResponseVO;
import com.example.common.config.AppConfig;
import com.example.common.constant.ConsoleColors;
import com.example.common.constant.Constants;
import com.example.common.enums.ConversationTypeEnum;
import com.example.common.enums.command.MessageCommand;
import com.example.common.model.ClientInfo;
import com.example.common.model.message.MessageContent;
import com.example.common.model.message.OfflineMessageContent;
import com.example.pack.message.ChatMessageAck;
import com.example.pack.message.MessageReciveServerAckPack;
import com.example.service.message.model.req.SendMessageReq;
import com.example.service.message.model.resp.SendMessageResp;
import com.example.service.seq.RedisSeq;
import com.example.service.utils.CallbackService;
import com.example.service.utils.ConversationIdGenerate;
import com.example.service.utils.MessageProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.Conversion;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@Slf4j
public class P2PMessageService {
    @Autowired
    CheckSendMessageService checkSendMessageService;

    @Autowired
    MessageProducer messageProducer;

    @Autowired
    RedisSeq redisSeq;

    @Autowired
    AppConfig appConfig;

    @Autowired
    CallbackService callbackService;

    @Autowired
    MessageStoreService messageStoreService;

    private static Logger logger = LoggerFactory.getLogger(P2PMessageService.class);

    //使用线程池进行提高消息处理的效率
    private final ThreadPoolExecutor threadPoolExecutor;

    {
        final AtomicInteger num = new AtomicInteger(0);
        threadPoolExecutor = new ThreadPoolExecutor(8, 8, 60, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(1000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("message-process-thread-" + num.getAndIncrement());
                return thread;
            }
        });
    }

    //发送消息的验证
    public void process(MessageContent messageContent){
        //前置校验
        //这个用户是否被禁言， 是否禁用
        //发送方是否是好友
        String fromId = messageContent.getFromId();
        String toId = messageContent.getToId();
        Integer appId=messageContent.getAppId();

        //从缓存中获取消息
        MessageContent messageFromMessageIdCache = messageStoreService.getMessageFromMessageIdCache(appId,messageContent.getMessageId(), MessageContent.class);
        if(messageFromMessageIdCache!=null){
            threadPoolExecutor.execute(()->{
                //1.回成功的ack给自己
                ack(messageContent);
                //2.发消息给同步在线端
                syncToSender(messageFromMessageIdCache,messageFromMessageIdCache);
                //3.发消息给对方在线端
                List<ClientInfo> clientInfos = dispatchMessage(messageContent);
                if (clientInfos.isEmpty()){
                    //发送接收确认给发送方,要带上是服务端发送的标示  在对方不在线的情况下
                    revicerAck(messageContent);
                }
            });
            return;
        }

        //发送消息之前的回调
        ResponseVO responseVO = ResponseVO.successResponse();
        if(appConfig.isSendMessageAfterCallback()){
            responseVO = callbackService.beforeCallback(messageContent.getAppId(), Constants.CallbackCommand.SendMessageBefore
                    , JSONObject.toJSONString(messageContent));
        }

        if(!responseVO.isOk()){
            ack(messageContent,responseVO);
            return;
        }


        //在进行消息保存前给消息一个标示 确保消息的有序性  格式为 appId + Seq + (fromId+toId) or groupId
        long seq=redisSeq.doGetSeq(messageContent.getAppId()+":"+Constants.SeqConstants.Message+ ":"+ConversationIdGenerate.generateP2PId(
                messageContent.getFromId(),messageContent.getToId()
        ));
        messageContent.setMessageSequence(seq);

//        ResponseVO responseVO = imServerPermissionCheck(fromId, toId, appId);
//        if (responseVO.isOk()){
        //使用线程池可能会引起 消息有序性的问题
            threadPoolExecutor.execute(()->{                               //使用线程池提高处理消息的效率

                messageStoreService.storeP2PMessage(messageContent);         //进行单聊的信息存储

                //插入离线消息
                OfflineMessageContent offlineMessageContent = new OfflineMessageContent();
                BeanUtils.copyProperties(messageContent,offlineMessageContent);
                offlineMessageContent.setConversationType(ConversationTypeEnum.P2P.getCode());
                messageStoreService.storeOfflineMessage(offlineMessageContent);


                //1.回成功的ack给自己
                ack(messageContent);
                //2.发消息给同步在线端
                syncToSender(messageContent,messageContent);
                //3.发消息给对方在线端
                List<ClientInfo> clientInfos = dispatchMessage(messageContent);

                //将消息的messageId进行保存到缓存中
                messageStoreService.setMessageFromMessageIdCache(appId,messageContent.getMessageId(),messageContent);


                if (clientInfos.isEmpty()){
                    //发送接收确认给发送方,要带上是服务端发送的标示  在对方不在线的情况下
                    revicerAck(messageContent);
                }

                //发送消息之后的回调
                if(appConfig.isSendMessageAfterCallback()){
                    callbackService.callback(messageContent.getAppId(),Constants.CallbackCommand.SendMessageAfter,
                            JSONObject.toJSONString(messageContent));
                }





            });
//        }
//        else {
//            //告诉客户端失败了
//            //ack
//            ack(messageContent,responseVO);
//        }
    }


    //ack是客户端返回给消息发送者的 看是消息发送成功还是失败
    private void ack(MessageContent messageContent){
//        logger.info(ConsoleColors.GREEN+"msg ack,msgId={},checkResut={}"+ConsoleColors.RESET,messageContent.getMessageId(),responseVO.getCode());

        ChatMessageAck chatMessageAck = new
                ChatMessageAck(messageContent.getMessageId(),messageContent.getMessageSequence());
        ResponseVO responseVO=new ResponseVO();
        responseVO.setData(chatMessageAck);
        //发消息
        messageProducer.sendToUser(messageContent.getFromId(), MessageCommand.MSG_ACK,
                responseVO,messageContent
        );
    }


    //这个ack是给回调的用的
    private void ack(MessageContent messageContent,ResponseVO responseVO){
        ChatMessageAck chatMessageAck = new
                ChatMessageAck(messageContent.getMessageId(),messageContent.getMessageSequence());
        responseVO.setData(chatMessageAck);
        //发消息
        messageProducer.sendToUser(messageContent.getFromId(), MessageCommand.MSG_ACK,
                responseVO,messageContent
        );
    }



    //进行确认接收消息
    public void revicerAck(MessageContent messageContent){
        MessageReciveServerAckPack pack = new MessageReciveServerAckPack();
        pack.setFromId(messageContent.getFromId());
        pack.setToId(messageContent.getFromId());
        pack.setMessageKey(messageContent.getMessageKey());
        pack.setMessageSequence(messageContent.getMessageSequence());
        pack.setServerSend(true);
        messageProducer.sendToUser(messageContent.getFromId(),MessageCommand.MSG_RECIVE_ACK,pack,new ClientInfo(messageContent.getAppId(),
                messageContent.getClientType(),messageContent.getImei()));

    }

    //进行发消息给其他端
    private void syncToSender(MessageContent messageContent, ClientInfo clientInfo){
        messageProducer.sendToUserExceptClient(messageContent.getFromId(),
                MessageCommand.MSG_P2P,messageContent,messageContent);
    }


    //发消息给 消息接收者的所有端
    private List<ClientInfo> dispatchMessage(MessageContent messageContent){
        List<ClientInfo> clientInfos = messageProducer.sendToUser(messageContent.getToId(), MessageCommand.MSG_P2P,
                messageContent, messageContent.getAppId());
        return clientInfos;
    }


    //判断禁言没，和好友关系的检查
    public ResponseVO imServerPermissionCheck(String fromId,String toId,
                                              Integer appId){
        ResponseVO responseVO = checkSendMessageService.checkSenderForvidAndMute(fromId, appId);
        if(!responseVO.isOk()){
            return responseVO;
        }
        responseVO = checkSendMessageService.checkFriendShip(fromId, toId, appId);
        return responseVO;
    }


    //消息的发送
    public SendMessageResp send(SendMessageReq req) {

        SendMessageResp sendMessageResp = new SendMessageResp();
        MessageContent message = new MessageContent();
        BeanUtils.copyProperties(req,message);
        //插入数据
        messageStoreService.storeP2PMessage(message);
        sendMessageResp.setMessageKey(message.getMessageKey());
        sendMessageResp.setMessageTime(System.currentTimeMillis());

        //2.发消息给同步在线端
        syncToSender(message,message);
        //3.发消息给对方在线端
        dispatchMessage(message);
        return sendMessageResp;
    }

}
